package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.util.MyUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Created by Smexy on 2022/11/25
 *
 *      每隔5分钟输出最近1小时内点击(pageView)量最多的前N个商品
 *              每隔5分钟输出: 计算要5min触发一次       slide
 *              最近1小时内：  计算的数据范围是最近1h内   size
 *
 *              基于eventtime的滑动时间窗口
 *
 *
 *       点击量最多的前N(3)个商品：
 *              ①统计过去窗口内，每一种商品的点击量   本质是窗口运算
 *                 输入:  一行是一个 Item 的 PV
     *                 举例：   111, a , pv
     *                         111, a , pv
     *                         111, b , pv
 *
 *                  运算:
 *                          按照itemId 分组
 *                          开一个1h大小，每次滑动5min的窗口
 *                          分组统计，每个item 出现的次数
 *
 *                  [8:55 ,9:55]   Task1
 *                           a:  10
 *                  [10:00,11:00]: Task1
 *                          a:  10
 *                          b:  11
 *
 *                  [10:00,11:00]： Task2
 *                          c:  12
 *                          d:  13
 *
 *              ②再对以上指定窗口内，所有商品的点击量进行排序，取前N
 *                  如果以上的结果都在一个Task，可以直接排序！
 *                  如果不在一个Task，需要按照窗口分组，把同一个窗口的合并到一起，再排序！
 *
 *                    [10:00,11:00]:
 *  *                          b:  11
 *  *                          c:  12
 *  *                          d:  13
 *
 *
 *                  排序，需要等算子的WM(时间)推进到所计算窗口的end时，再执行排序。
 *                          在规定的时间做时间，使用定基于eventTime的定时器实现！
 *
 *
 */
public class Demo3_TopN
{
    public static void main(String[] args) {


       Configuration conf = new Configuration();
       conf.setInteger("rest.port", 3333);
       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
            .withTimestampAssigner((u, t) -> u.getTimestamp() * 1000);

        env.readTextFile("data/UserBehavior.csv")
           .flatMap(new FlatMapFunction<String, UserBehavior>()
           {
               @Override
               public void flatMap(String value, Collector<UserBehavior> out) throws Exception {
                   String[] words = value.split(",");
                   if ("pv".equals(words[3])){
                       out.collect(new UserBehavior(
                           Long.valueOf(words[0]),
                           Long.valueOf(words[1]),
                           Integer.valueOf(words[2]),
                           words[3],
                           Long.valueOf(words[4])
                       ));
                   }
               }
           })
           .assignTimestampsAndWatermarks(watermarkStrategy)
            //第一阶段:从 UserBehavior  转为   TopN
            .keyBy(UserBehavior::getItemId)
            .window(SlidingEventTimeWindows.of(Time.hours(1),Time.minutes(5)))
            .aggregate(new AggregateFunction<UserBehavior, TopN, TopN>()
            {
                @Override
                public TopN createAccumulator() {
                    return new TopN(0l, 0l, "a", 0);
                }

                @Override
                public TopN add(UserBehavior value, TopN acc) {
                    acc.setCount(acc.getCount() + 1);
                    return acc;
                }

                @Override
                public TopN getResult(TopN acc) {
                    return acc;
                }

                @Override
                public TopN merge(TopN a, TopN b) {
                    return null;
                }
            }, new ProcessWindowFunction<TopN, TopN, Long, TimeWindow>()
            {
                @Override
                public void process(Long key, Context context, Iterable<TopN> elements, Collector<TopN> out) throws Exception {
                    TopN topN = elements.iterator().next();
                    //赋予窗口和Key
                    TimeWindow window = context.window();
                    topN.setStart(window.getStart());
                    topN.setEnd(window.getEnd());
                    topN.setItemId(key.toString());

                    //输出
                    out.collect(topN);

                }
            })
          //第二阶段: 按照窗口id，分组，分组后在下游使用定时器排序
           .keyBy(TopN::getStart)
           .process(new KeyedProcessFunction<Long, TopN, String>()
           {

               //判断是不是要计算的窗口的一个到达的数据
               private ValueState<Boolean> first;
               private ListState<TopN> topN;

               @Override
               public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        //对ListState<TopN> topN排序取前3
                   List<TopN> list = StreamSupport.stream(topN.get().spliterator(), true)
                                                     .sorted((t1, t2) -> -t1.getCount().compareTo(t2.getCount()))
                                                     .limit(3)
                                                     .collect(Collectors.toList());


                      //生成最终结果
                   TopN topN = list.get(0);

                   TimeWindow timeWindow = new TimeWindow(topN.getStart(), topN.getEnd());

                   String top3Str = list.stream().map(t -> t.getItemId() + ":" + t.getCount()).collect(Collectors.joining(","));

                   out.collect(MyUtil.parseTimeWindow(timeWindow) +" 的top3: "+ top3Str);

               }

               @Override
               public void open(Configuration parameters) throws Exception {
                   topN = getRuntimeContext().getListState(new ListStateDescriptor<TopN>("topN", TopN.class));

                   first = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("first", Boolean.class));
               }

               //来一个调用一次
               // 如果是开窗，你传入的是  ProcessWindowFunction，其中的 processElement 是一个窗口执行一次
               // 如果不开窗，你传入的是  KeyedProcessFunction，其中的 processElement 是一个元素执行一次
               @Override
               public void processElement(TopN value, Context ctx, Collector<String> out) throws Exception {
                   if (first.value() == null){
                       //注册定时器,只执行一次
                       ctx.timerService().registerEventTimeTimer(value.end + 5000);
                       first.update(true);
                   }
                    //来一个元素加入到 ListState<TopN> topN
                   topN.add(value);

               }
           }).setParallelism(1)
           .print().setParallelism(1);

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }


    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class TopN{
        //[start,end]
        private Long start;
        private Long end;
        private String itemId;
        private Integer count;

    }
}
